package spark

import fun.functions.{getKafkaData, openSpark, sendKafka}


object streamKafka1 {
  def main(args: Array[String]): Unit = {
    val streamingContext=openSpark
    val lines=getKafkaData(streamingContext)

    // 更新状态
    val updateGenderCount = (values: Seq[Int], state: Option[Int]) => {
      val currentCount = values.sum
      val previousCount = state.getOrElse(0)
      Some(currentCount + previousCount)
    }

    /**
     * 每隔2秒分别统计所有男女人数各自的总和 result
     * */
    val genderCount = lines
      .map { line =>
        val gender = line.split("\t")(2).toInt
        (gender, 1)
      }
      .updateStateByKey(updateGenderCount)


    genderCount.foreachRDD { rdd =>
      val genderCountList = rdd.collect().toList
      val record: StringBuilder = new StringBuilder("{")
      genderCountList.foreach {
        case (gender, count) =>
          record.append(s""""$gender":$count,""")
      }
      record.setLength(record.length-1)
      record.append("}")
      sendKafka("result", record)
    }

    streamingContext.start()
    streamingContext.awaitTermination()

  }

}